package com.young.trigger.handler;

import com.young.scheduler.grpc.SchedulerModel;
import com.young.scheduler.grpc.VClientSheduleServiceGrpc;
import com.young.trigger.configuration.WebProperties;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/3/24 10:38 AM
 */
@Component
public class GrpcClient implements Closeable {

    @Autowired
    private WebProperties webProperties;

    private static final int timeout = 3000;

    private final Map<String,Client> pool = new ConcurrentHashMap<>();

    private static final Logger LOGGER = LoggerFactory.getLogger(GrpcClient.class);


    public int call(HttpRequest request) {

        try {
            Client client = get(request.selected().getIp(),request.selected().getPort());

            client.open();


            VClientSheduleServiceGrpc.VClientSheduleServiceBlockingStub service =
                    VClientSheduleServiceGrpc.newBlockingStub(client.channel).withDeadlineAfter(timeout, TimeUnit.MILLISECONDS);

            SchedulerModel.JobRequest.Builder builder = SchedulerModel.JobRequest
                    .newBuilder().setJobUniqCode(request.job.getUniqueCode())
                    .setJobArgs(request.job.getData())
                    .setTraceId(request.traceId)
                    .setJobRunStrategy(request.job.getStrategy())
                    .setJobNodeNum(String.valueOf(request.total))
                    .setJobNodeIndex(String.valueOf(request.index));

            builder.addAllCallbackNodes(BrokerContext.getBrokers());
            builder.setCallbackNodePort(webProperties.getPort() == null ? 80 : webProperties.getPort());

            SchedulerModel.JobResponse response = service.doJob(builder.build());

            LOGGER.info("GrpcClient - grpc请求结果,traceId - {},status - {},message - {}",
                    request.traceId, response.getStatus(),response.getMsg());

            return response.getStatus();
        } catch (Exception e) {
            LOGGER.info("GrpcClient - grpc请求失败,traceId - {},{}/{}.{},client - [{}:{}],异常 - {} : {}",
                    request.traceId,
                    request.job.getProject(),
                    request.job.getClazz(),
                    request.job.getMethod(),
                    request.selected().getIp(),
                    request.selected().getPort(),
                    e.getClass().getName(),
                    e.getMessage());
        }
        return -2;
    }

    private Client get(String host,int port) {
        String key = host + "_" + port;
        if (pool.containsKey(key)) {
            return pool.get(key);
        }

        synchronized (pool) {
            if (pool.containsKey(key)) {
                return pool.get(key);
            }
            Client client = new Client(host,port);
            pool.put(key,client);
            return client;
        }

    }

    @Override
    public void close() throws IOException {
        pool.forEach((key,value) -> {
            try {
                value.close();
            } catch (Exception e) {
                //
            }
        });
    }


    public class Client implements Closeable {

        private ManagedChannel channel;
        private String host;
        private int port;

        public Client(String host, int port) {
            channel = ManagedChannelBuilder.forAddress(host, port)
                    .usePlaintext(true)
                    .build();
            this.host = host;
            this.port = port;
        }

        private void open() {
            if (this.channel.isTerminated() || this.channel.isShutdown()) {
                synchronized (this) {
                    if (this.channel.isTerminated() || this.channel.isShutdown()) {
                        channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext(true).build();
                    }
                }
            }
        }

        @Override
        public void close() {
            channel.shutdown();
        }
    }
}
